package drds.server.execute_engine.data_handling;

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import drds.server.Server;
import drds.server.execute_engine.data_handling.tmp.RowDataSorter;
import drds.server.handler.MultiNodeQueryHandler;
import drds.server.net.packet.ByteBufferUtil;
import drds.server.net.packet.EofPacket;
import drds.server.net.packet.RowDataPacket;
import drds.server.route.Route;
import drds.server.route.RouteNode;
import drds.server.server.SessionContext;

/**
 * Data merge service handle data Min,Max,AVG group 、order by 、limit
 * 
 * /modify by coder_czp/2015/11/2
 * 
 * Fixbug: sql timeout and hang problem.
 * 
 * 
 */
public class DataMergeService extends AbstractDataNodeMerge {

	private RowDataSorter rowDataSorter;
	private RowDataPacketGrouper rowDataPacketGrouper;
	private Map<String, LinkedList<RowDataPacket>> result = new HashMap<String, LinkedList<RowDataPacket>>();
	private ConcurrentHashMap<String, Boolean> canDiscard = new ConcurrentHashMap<String, Boolean>();

	public DataMergeService(MultiNodeQueryHandler handler, Route route) {
		super(handler, route);
		for (RouteNode node : route.getRouteNodes()) {
			result.put(node.getName(), new LinkedList<RowDataPacket>());
		}
	}

	/**
	 * @param columMap
	 * @param fieldCount
	 */
	public void onRowMetaData(Map<String, Column> columMap, int fieldCount) {

		int[] groupColumnIndexs = null;
		this.fieldCount = fieldCount;

		if (route.getGroupByColumns() != null) {

			groupColumnIndexs = toColumnIndex(route.getGroupByColumns(), columMap);
		}

		if (route.getGroupByFilter() != null) {
			Column colMeta = columMap.get(route.getGroupByFilter().getLeft().toUpperCase());
			if (colMeta != null) {
				route.getGroupByFilter().setColumn(colMeta);
			}
		}

		if (route.isHasAggrColumn()) {
			List<AggregateFunctionType> mergCols = new LinkedList<AggregateFunctionType>();
			Map<String, Integer> mergeColsMap = route.getMergeColumnAndIndexMap();

			if (mergeColsMap != null) {
				for (Map.Entry<String, Integer> mergEntry : mergeColsMap.entrySet()) {
					String colName = mergEntry.getKey().toUpperCase();
					int type = mergEntry.getValue();
					if (AggregateFunctionType.AVG == type) {

						Column sumColMeta = columMap.get(colName + "SUM");
						Column countColMeta = columMap.get(colName + "COUNT");
						if (sumColMeta != null && countColMeta != null) {
							Column colMeta = new Column(sumColMeta.columIndex, countColMeta.columIndex, sumColMeta.getColumType());
							mergCols.add(new AggregateFunctionType(colMeta, mergEntry.getValue()));
						}
					} else {

						Column colMeta = columMap.get(colName);
						mergCols.add(new AggregateFunctionType(colMeta, mergEntry.getValue()));
					}
				}
			}
			// add no alias merg column
			for (Map.Entry<String, Column> fieldEntry : columMap.entrySet()) {
				String colName = fieldEntry.getKey();
				int result = AggregateFunctionType.tryParseAggCol(colName);
				if (result != AggregateFunctionType.UNSUPPORT && result != AggregateFunctionType.NOMERGE) {
					mergCols.add(new AggregateFunctionType(fieldEntry.getValue(), result));
				}
			}

			rowDataPacketGrouper = new RowDataPacketGrouper(groupColumnIndexs, mergCols.toArray(new AggregateFunctionType[mergCols.size()]), route.getGroupByFilter());
		}

		if (route.getOrderByColumnMap() != null) {
			LinkedHashMap<String, Integer> orders = route.getOrderByColumnMap();
			OrderByColum[] orderCols = new OrderByColum[orders.size()];
			int i = 0;
			for (Map.Entry<String, Integer> entry : orders.entrySet()) {
				String key = entry.getKey().toUpperCase();
				Column colMeta = columMap.get(key);
				if (colMeta == null) {
					throw new IllegalArgumentException("all columns in order by clause should be in the selected column list!" + entry.getKey());
				}
				orderCols[i++] = new OrderByColum(colMeta, entry.getValue());
			}

			RowDataSorter tmp = new RowDataSorter(orderCols);
			tmp.setLimit(route.getLimitStart(), route.getLimitSize());
			rowDataSorter = tmp;
		}

		if (Server.getInstance().getConfig().getDRDS().getUseStreamOutput() == 1 && rowDataPacketGrouper == null && rowDataSorter == null) {
			setStreamOutputResult(true);
		} else {
			setStreamOutputResult(false);
		}
	}

	/**
	 * release resources
	 */
	public void clear() {
		result.clear();
		rowDataPacketGrouper = null;
		rowDataSorter = null;
	}

	@Override
	public void run() {
		// sort-or-group: no need for us to using multi-threads, because
		// both sorter and group are synchronized!!
		// @author Uncle-pan
		// @since 2016-03-23
		if (!running.compareAndSet(false, true)) {
			return;
		}
		// eof handler has been placed to "if (pack == END_FLAG_PACK){}" in
		// for-statement
		// @author Uncle-pan
		// @since 2016-03-23
		boolean nulpack = false;
		try {
			// loop-on-packs
			for (;;) {
				final PackWraper pack = packWraperQueue.poll();
				// async: handling row pack queue, this business thread should
				// exit when no pack
				// @author Uncle-pan
				// @since 2016-03-23
				if (pack == null) {
					nulpack = true;
					break;
				}
				// eof: handling eof pack and exit
				if (pack == END_FLAG_PACK) {

					final int warningCount = 0;
					final EofPacket eofp = new EofPacket();
					final ByteBuffer eof = ByteBuffer.allocate(9);
					ByteBufferUtil.writeInt3(eof, eofp.computePacketSize());
					eof.put(eofp.packetId);
					eof.put(eofp.fieldCount);
					ByteBufferUtil.writeInt2(eof, warningCount);
					ByteBufferUtil.writeInt2(eof, eofp.status);
					final SessionContext source = multiQueryHandler.getSession().getSessionContext();
					final byte[] array = eof.array();
					multiQueryHandler.outputMergeResult(source, array, getRowDataPacketList(array));
					break;
				}

				// merge: sort-or-group, or simple add
				final RowDataPacket row = new RowDataPacket(fieldCount);
				row.read(pack.rowData);

				if (rowDataPacketGrouper != null) {
					rowDataPacketGrouper.addRowDataPacket(row);
				} else if (rowDataSorter != null) {
					if (!rowDataSorter.addRowDataPacket(row)) {
						canDiscard.put(pack.dataNode, true);
					}
				} else {
					result.get(pack.dataNode).add(row);
				}
			}// rof
		} catch (final Exception e) {
			multiQueryHandler.handleDataProcessException(e);
		} finally {
			running.set(false);
		}
		// try to check packs, it's possible that adding a pack after polling a
		// null pack
		// and before this time pointer!!
		// @author Uncle-pan
		// @since 2016-03-23
		if (nulpack && !packWraperQueue.isEmpty()) {
			this.run();
		}
	}

	/**
	 * return merged data
	 * 
	 * @return (最多i*(offset+size)行数据)
	 */
	public List<RowDataPacket> getRowDataPacketList(byte[] eof) {

		List<RowDataPacket> tempRowDataPacketList = null;

		if (this.rowDataPacketGrouper != null) {
			tempRowDataPacketList = rowDataPacketGrouper.getRowDataPacketList();
			rowDataPacketGrouper = null;
		}

		if (rowDataSorter != null) {

			if (tempRowDataPacketList != null) {
				Iterator<RowDataPacket> iterator = tempRowDataPacketList.iterator();
				while (iterator.hasNext()) {
					rowDataSorter.addRowDataPacket(iterator.next());
					iterator.remove();
				}
			}
			tempRowDataPacketList = rowDataSorter.getSortedRowDataPacketList();
			rowDataSorter = null;
		}

		// no grouper and sorter
		if (tempRowDataPacketList == null) {
			tempRowDataPacketList = new LinkedList<RowDataPacket>();
			for (RouteNode routeNode : route.getRouteNodes()) {
				tempRowDataPacketList.addAll(result.get(routeNode.getName()));
			}
		}

		return tempRowDataPacketList;
	}
}
